package com.ke.hbase;

import com.ke.PhoneInfo;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;

public class HBaseDemo {

    Configuration conf = null;
    Connection conn = null;
    // 表的管理对象
    Admin admin = null;
    Table table = null;
    // 创建表的对象
    TableName tableName = TableName.valueOf("number3");

    @Before
    public void init() throws IOException {
        // 创建配置文件对象
        conf = HBaseConfiguration.create();
        // 加载zookeeper配置
        conf.set("hbase.zookeeper.quorum","ke02,ke03,ke04");
        // 获取连接
        conn = ConnectionFactory.createConnection(conf);
        // 获取对象
        admin = conn.getAdmin();
        // 获取数据操作对象
        table = conn.getTable(tableName);
    }

    @Test
    public void createTable() throws IOException{
        // 定义表描述对象
        TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tableName);
        // 定义列族对象
        ColumnFamilyDescriptorBuilder columnFamilyDescriptorBuilder = ColumnFamilyDescriptorBuilder.newBuilder("cf".getBytes());
        // 添加列族信息给表
        tableDescriptorBuilder.setColumnFamily(columnFamilyDescriptorBuilder.build());

        if(admin.tableExists(tableName)){
            //禁用表
            admin.disableTable(tableName);
            admin.deleteTable(tableName);
        }
        //创建表
        admin.createTable(tableDescriptorBuilder.build());
    }


    @Test
    public void insert() throws IOException {
        Put put = new Put(Bytes.toBytes("row_two"));
        put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("name"), Bytes.toBytes("lisi"));
        put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("age"), Bytes.toBytes("341"));
        put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("sex"), Bytes.toBytes("women"));
        table.put(put);
    }


    /**
     * Hbase 可以插入十亿行 和百万列, 所以查询需要克服io，如下减少io
     * 查看一行row
     * @throws IOException
     */
    @Test
    public void get() throws IOException{

        Get get = new Get(Bytes.toBytes("row_two"));

        // 在服务端做数据过滤，挑选出符合需求的列,也就是百万列，我只要这三个，其他的不查也就不走io
        get.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("name"));
        get.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("age"));
        get.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("sex"));

        Result result = table.get(get);
        Cell cell1 = result.getColumnLatestCell(Bytes.toBytes("cf"), Bytes.toBytes("name"));
        Cell cell2 = result.getColumnLatestCell(Bytes.toBytes("cf"), Bytes.toBytes("age"));
        Cell cell3 = result.getColumnLatestCell(Bytes.toBytes("cf"), Bytes.toBytes("sex"));

        String name = Bytes.toString(CellUtil.cloneValue(cell1));
        String age = Bytes.toString(CellUtil.cloneValue(cell2));
        String sex = Bytes.toString(CellUtil.cloneValue(cell3));

        System.out.println(name);
        System.out.println(age);
        System.out.println(sex);
    }

    /**
     * Hbase 可以插入十亿行 和百万列, 所以查询需要克服io，如下减少io
     * 查看所有行row, 公司一般禁止使用，相当于mysql的select * from
     * @throws IOException
     */
    @Test
    public void scan() throws IOException{

        Scan scan = new Scan();
        ResultScanner scanner = table.getScanner(scan);
        for (Result result : scanner) {
            Cell cell1 = result.getColumnLatestCell(Bytes.toBytes("cf"), Bytes.toBytes("name"));
            Cell cell2 = result.getColumnLatestCell(Bytes.toBytes("cf"), Bytes.toBytes("age"));
            Cell cell3 = result.getColumnLatestCell(Bytes.toBytes("cf"), Bytes.toBytes("sex"));

            String name = Bytes.toString(CellUtil.cloneValue(cell1));
            String age = Bytes.toString(CellUtil.cloneValue(cell2));
            String sex = Bytes.toString(CellUtil.cloneValue(cell3));
            System.out.println(name);
            System.out.println(age);
            System.out.println(sex);
        }
    }


    @Test
    public void insertMangerData() throws Exception{
        List<Put> puts = new ArrayList<Put>();
        for (int i = 0; i < 10; i++) {
            String myNumber = getNumber("177");
            for (int j = 0; j < 10; j++) {
                // 模拟数据 别人的手机号码、通话时长、通话时间、通话类型
                String otherNumber = getNumber("177");
                String length = String.valueOf(random.nextInt(100));
                String date = getDate("2019");
                String type = String.valueOf(random.nextInt(2));

                // rowkey 设计为我的手机号码 + 时间戳反转(也可以将手机号码倒过来进行数据分散)
                String rowkey = myNumber+"_"+(Long.MAX_VALUE-sdf.parse(date).getTime());
                Put put = new Put(Bytes.toBytes(rowkey));
                put.addColumn(Bytes.toBytes("cf"),Bytes.toBytes("otherNumber"),Bytes.toBytes(otherNumber));
                put.addColumn(Bytes.toBytes("cf"),Bytes.toBytes("length"),Bytes.toBytes(length));
                put.addColumn(Bytes.toBytes("cf"),Bytes.toBytes("date"),Bytes.toBytes(date));
                put.addColumn(Bytes.toBytes("cf"),Bytes.toBytes("type"),Bytes.toBytes(type));
                puts.add(put);
            }
        }
        table.put(puts);
    }

    /**
     * 查询某一个用户3月份的通话记录
     */
    @Test
    public void scanByCondition2() throws Exception {
        Scan scan = new Scan();
        String startRow = "17782779334_"+(Long.MAX_VALUE-sdf.parse("20190331000000").getTime());
        String stopRow = "17782779334_"+(Long.MAX_VALUE-sdf.parse("20190301000000").getTime());
        scan.withStartRow(Bytes.toBytes(startRow));
        scan.withStopRow(Bytes.toBytes(stopRow));
        ResultScanner rss = table.getScanner(scan);
        for (Result result:rss) {
            Cell cell1 = result.getColumnLatestCell(Bytes.toBytes("cf"), Bytes.toBytes("otherNumber"));
            Cell cell2 = result.getColumnLatestCell(Bytes.toBytes("cf"), Bytes.toBytes("length"));
            Cell cell3 = result.getColumnLatestCell(Bytes.toBytes("cf"), Bytes.toBytes("type"));
            Cell cell4 = result.getColumnLatestCell(Bytes.toBytes("cf"), Bytes.toBytes("date"));

            String otherNumber = Bytes.toString(CellUtil.cloneValue(cell1));
            String length = Bytes.toString(CellUtil.cloneValue(cell2));
            String type = Bytes.toString(CellUtil.cloneValue(cell3));
            String date = Bytes.toString(CellUtil.cloneValue(cell4));

            System.out.print(otherNumber + " - ");
            System.out.print(length + " - ");
            System.out.print(type + " - ");
            System.out.println(date);
        }
    }



    @Test
    public void scan_number() throws IOException{

        Scan scan = new Scan();
        ResultScanner scanner = table.getScanner(scan);
        for (Result result : scanner) {
            Cell cell1 = result.getColumnLatestCell(Bytes.toBytes("cf"), Bytes.toBytes("otherNumber"));
            Cell cell2 = result.getColumnLatestCell(Bytes.toBytes("cf"), Bytes.toBytes("length"));
            Cell cell3 = result.getColumnLatestCell(Bytes.toBytes("cf"), Bytes.toBytes("type"));
            Cell cell4 = result.getColumnLatestCell(Bytes.toBytes("cf"), Bytes.toBytes("date"));

            String otherNumber = Bytes.toString(CellUtil.cloneValue(cell1));
            String length = Bytes.toString(CellUtil.cloneValue(cell2));
            String type = Bytes.toString(CellUtil.cloneValue(cell3));
            String date = Bytes.toString(CellUtil.cloneValue(cell4));

            System.out.println(otherNumber + " - - " + length + " - - " + type + " - - " + date);

        }
    }


    @Test
    public void delete() throws Exception {
        Delete delete = new Delete("17782779334_9223370490661937807".getBytes());
        table.delete(delete);
    }


    /**
     * 序列化数据: 可以减少内存、存储的数据量大小
     * @throws Exception
     */
    @Test
    public void insertMangerByProtoBufData() throws Exception{
        List<Put> puts = new ArrayList<Put>();
        for (int i = 0; i < 10; i++) {
            String myNumber = getNumber("177");
            for (int j = 0; j < 10; j++) {
                // 模拟数据 别人的手机号码、通话时长、通话时间、通话类型
                String otherNumber = getNumber("177");
                String length = String.valueOf(random.nextInt(100));
                String date = getDate("2019");
                String type = String.valueOf(random.nextInt(2));

                // rowkey 设计为我的手机号码 + 时间戳反转(也可以将手机号码倒过来进行数据分散)
                String rowkey = myNumber+"_"+(Long.MAX_VALUE-sdf.parse(date).getTime());

                PhoneInfo.PhoneDetail.Builder builder = PhoneInfo.PhoneDetail.newBuilder();
                builder.setDate(date);
                builder.setLength(length);
                builder.setType(type);
                builder.setOtherNumber(otherNumber);
                Put put = new Put(Bytes.toBytes(rowkey));
                // 序列化后就没有列名了，所以需要自定义列名
                put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("phone"), builder.build().toByteArray());
                puts.add(put);
            }
        }
        table.put(puts);
        /*
        HBase中序列化数据：
17786774306_9223370490639607807                             column=cf:phone, timestamp=1605756025604, value=\x0A\x0B17714285591\x12\x0241\x1A\x010"\x0E20190100081248
17786774306_9223370490642379807                                     column=cf:phone, timestamp=1605756025604, value=\x0A\x0B17794212043\x12\x0274\x1A\x010"\x0E20190100072636
         */
    }

    /**
     * 查看数据，需要反序列化
     * @throws Exception
     */
    @Test
    public void getByProtoBuf() throws Exception {
        Get get = new Get("17789707849_9223370473395733807".getBytes());
        Result result = table.get(get);
        byte[] info = CellUtil.cloneValue(result.getColumnLatestCell(Bytes.toBytes("cf"), Bytes.toBytes("phone")));
        PhoneInfo.PhoneDetail phoneDetail = PhoneInfo.PhoneDetail.parseFrom(info);
        System.out.println(phoneDetail);
        /*
        结果:
        otherNumber: "17714285591"
        length: "41"
        type: "0"
        date: "20190100081248"
         */
    }




    SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");
    private String getDate(String s) {
        return s+String.format("%02d%02d%02d%02d%02d",random.nextInt(12)+1,random.nextInt(31),random.nextInt(24),random.nextInt(60),random.nextInt(60));
    }

    Random random = new Random();
    private String getNumber(String s) {
        return s + String.format("%08d", random.nextInt(99999999));
    }


    @After
    public void destory(){
        try {
            table.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
        try {
            admin.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
        try {
            conn.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }



}
